package cn.itcast.dmp.runner

import cn.itcast.dmp.area.AreaProcessor
import cn.itcast.dmp.config.AppConfigHelper
import cn.itcast.dmp.untils.SparkSessionUtils
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, SparkSession}

object BusinessAreasRunner2 {
  private val SOURCE_TABLE_NAME = AppConfigHelper.AD_MAIN_TABLE_NAME
  private val SINK_TABLE_NAME = AppConfigHelper.BUSINESS_AREAS_TABLE_NAME

  def main(args: Array[String]): Unit = {
    // 第一、创建SparkSession
    val sparkSession = SparkSessionUtils.createSparkSession(this.getClass)
    // 第二、加载ods表的数据
    import cn.itcast.dmp.untils.KuduUtils._
    val odsDf:DataFrame = sparkSession.readKuduTable(SOURCE_TABLE_NAME) match{

        //进行判断，如果有数据那么久获取，没有则返回
      case Some(df) => df
      case None => println("error :ods 表无数据"); return
    }
    // 第三、获取（新的）商圈数据
    val areaDF:DataFrame = AreaProcessor.processData(odsDf)
    // 第四、将处理转换的数据DataFrame保存到Kudu表中
    import cn.itcast.dmp.untils.KuduUtils._
    // a. 先创建表，主键为geoHash
    sparkSession.createKuduTable(SINK_TABLE_NAME,areaDF.schema,Seq("geo_hash"),isDelete = false)
    // b. 保存数据
    areaDF.saveAsKuduTable(SINK_TABLE_NAME)
    // 第五、应用结束关闭资源
    sparkSession.stop()

  }

}
